
(import '[org.apache.kafka.clients.producer KafkaProducer])
(import '[org.apache.kafka.clients.consumer KafkaConsumer])
(import '[java.util Properties])
(import '[org.apache.kafka.clients.producer ProducerRecord])

(def env "nexus_test")
(def kafkas "10.25.21.110:9092")
(def group "larluo")

(def in-props 
  (doto (Properties.)
    (.put "bootstrap.servers" kafkas)
    (.put "auto.offset.reset" "earliest")
    (.put "group.id" group)
    (.put "key.deserializer" "org.apache.kafka.common.serialization.ByteArrayDeserializer")
    (.put "value.deserializer" "org.apache.kafka.common.serialization.ByteArrayDeserializer")
    ))

(def out-props
  (doto (Properties.)
    (.put "bootstrap.servers" kafkas)
    (.put "compression.type" "zstd")
    (.put "key.serializer" "org.apache.kafka.common.serialization.ByteArraySerializer")
    (.put "value.serializer" "org.apache.kafka.common.serialization.ByteArraySerializer")
    )
  )
  #_(do
    (def consumer (KafkaConsumer. in-props))
    (def duration (java.time.Duration/ofMillis Long/MAX_VALUE))
    (def topic (format "data_buffer_%s_dc_sdk_push_json", env))
    (println "subscribe topic")
    (.subscribe consumer [topic])
    (println "consume record")
    (doseq [record (.poll consumer duration)]
      (println "record:" record))
    )
  (do
    (def producer (KafkaProducer. out-props))
    (def topic (format "data_buffer_%s_dc_sdk_push_avro", env))
    (.send producer (ProducerRecord. topic nil (.getBytes "larluo")))
    )
